package com.ktvme.controller;

import com.ktvme.core.*;
import com.ktvme.entity.Cluster;
import com.ktvme.flink13.Flink1_13JobServiceImpl;
import com.ktvme.result.Result;
import com.ktvme.service.ClusterService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.io.File;
import java.util.List;

@RestController
@RequestMapping("/BDMobieOps/flink")
@Slf4j
@Tag(name = "Flink 接口")
public class FlinkController {

    @Autowired
    private ClusterService clusterService;

    private AbstractFlinkJobService getFlinkClient(String clusterId) {
        Cluster cluster = clusterService.getById(clusterId);
        if (cluster == null) {
            throw new IllegalArgumentException("Cluster not found with id: " + clusterId);
        }
        return FlinkJobServiceFactory.createFlinkJobService(cluster.getVersion(), cluster.getUrl());
    }

    @Operation(summary = "获取所有作业")
    @GetMapping("/{clusterId}/jobs")
    public Result<List<JobInfo>> getAllJobs(@PathVariable("clusterId") String clusterId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JobInfo> runningJobs = flinkClient.getRunningJobs();
        List<JobInfo> finishedJobs = flinkClient.getFinishedJobs();
        runningJobs.addAll(finishedJobs);
        return Result.success(runningJobs);
    }

    @Operation(summary = "获取运行中的作业")
    @GetMapping("/{clusterId}/jobs/running")
    public Result<List<JobInfo>> getRunningJobs(@PathVariable("clusterId") String clusterId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JobInfo> runningJobs = flinkClient.getRunningJobs();
        return Result.success(runningJobs);
    }

    @Operation(summary = "获取已完成的作业")
    @GetMapping("/{clusterId}/jobs/finished")
    public Result<List<JobInfo>> getFinishedJobs(@PathVariable("clusterId") String clusterId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JobInfo> finishedJobs = flinkClient.getFinishedJobs();
        return Result.success(finishedJobs);
    }

    @Operation(summary = "获取作业详情")
    @GetMapping("/{clusterId}/jobs/{jobId}")
    public Result<JobInfo> getJobInfo(
            @PathVariable("clusterId") String clusterId,
            @PathVariable("jobId") String jobId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        JobInfo jobInfo = flinkClient.getJobInfo(jobId);
        return Result.success(jobInfo);
    }

    @Operation(summary = "获取已上传的JAR包列表")
    @GetMapping("/{clusterId}/jars")
    public Result<List<JarInfo>> listJars(@PathVariable("clusterId") String clusterId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JarInfo> jars = flinkClient.listJars();
        return Result.success(jars);
    }

    @Operation(summary = "上传JAR包")
    @PostMapping("/{clusterId}/jars/upload")
    public Result<JarInfo> uploadJar(
            @PathVariable("clusterId") String clusterId,
            @RequestParam("file") MultipartFile file) {
        try {
            // 将 MultipartFile 转换为临时文件
            File tempFile = File.createTempFile("flink_", ".jar");
            file.transferTo(tempFile);

            AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
            JarInfo jarInfo = flinkClient.uploadJar(tempFile);

            // 删除临时文件
            tempFile.delete();

            return Result.success(jarInfo);
        } catch (Exception e) {
            log.error("上传JAR包失败", e);
            return Result.error("上传JAR包失败: " + e.getMessage());
        }
    }

    @Operation(summary = "删除JAR包")
    @DeleteMapping("/{clusterId}/jars/{jarId}")
    public Result deleteJar(
            @PathVariable("clusterId") String clusterId,
            @PathVariable("jarId") String jarId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        flinkClient.deleteJar(jarId);
        return Result.success();
    }

    @Operation(summary = "提交作业")
    @PostMapping("/{clusterId}/jobs/run")
    public Result<String> runJob(
            @PathVariable("clusterId") String clusterId,
            @RequestBody RunJarRequest request) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        String jobId = flinkClient.runJar(request);
        return Result.success(jobId);
    }

    @Operation(summary = "取消作业")
    @PostMapping("/{clusterId}/jobs/{jobId}/cancel")
    public Result cancelJob(
            @PathVariable("clusterId") String clusterId,
            @PathVariable("jobId") String jobId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        flinkClient.cancelJob(jobId);
        return Result.success();
    }

    @Operation(summary = "重启失败的作业")
    @PostMapping("/{clusterId}/jobs/{jobId}/restart")
    public Result restartFailedJob(
            @PathVariable("clusterId") String clusterId,
            @PathVariable("jobId") String jobId) {
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        flinkClient.restartFailedJob(jobId);
        return Result.success();
    }

} 